package com.etlmaster.util.vfs;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class HdfsUtils {
	private static Logger LOG = LoggerFactory.getLogger(HdfsUtils.class);
	
	public static FileSystem getFileSystem() {
		return getFileSystem(null);
	}
	
	public static FileSystem getFileSystem(String clusterName) {
		return getFileSystem(clusterName, 0);
	}
	
	public static FileSystem getFileSystem(String hostName,int port) {
		Configuration conf = new Configuration();
		String clusterName = hostName;
		if(clusterName!=null && clusterName.length()>0){
			ResourcePatternResolver resourcePatternResolver = new PathMatchingResourcePatternResolver();
			String packageSearchPath = "classpath*:conf/datax/"+clusterName+"-*-site.xml";
			try {
				Resource[] resources = resourcePatternResolver.getResources(packageSearchPath);
				if(resources.length>0){
					for (Resource resource : resources) {
						conf.addResource(resource.getFilename());
					}
					LOG.info("加载[{}]集群配置",clusterName);
				} else {
					String url = clusterName;
					
					if(!url.startsWith("hdfs://")){
						url = "hdfs://" + clusterName;
					}
					
					if(port>0){
						url = url+ ":"+port;
					}
					
					LOG.info("未找到[{}]集群配置，加载URL：{}",clusterName,url);
					conf.set("fs.defaultFS", url);
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		try {
			return FileSystem.get(conf);
		} catch (Exception e) {
			LOG.error(e.getMessage(), e);
		}
		return null;
	}

	public static void closeHdfs(FileSystem fs) {
		try {
			fs.close();
			LOG.info("Hdfs connection closed");
		} catch (IOException e) {
			LOG.warn(e.getMessage());
		}
	}
}
